package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo10Join {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("join").setMaster("local")

    val sc = new SparkContext(conf)

    /**
      * join：默认时内连接
      *
      * 通过key进行关联
      *
      */
    //读取students数据
    val students: RDD[String] = sc.textFile("data/students.txt")

    //读取score数据
    val scores: RDD[String] = sc.textFile("data/score.txt")

    //将RDD转化成kv格式
    val stuKvRDD: RDD[(String, String)] = students.map(student => {
      val split: Array[String] = student.split(",")
      //学号作为key，学生信息作为value
        //学号
      val id: String = split(0)
      (id,student)
    })

    val scoKvRDD: RDD[(String, String)] = scores.map(score =>{
      val split: Array[String] = score.split(",")
      //学号
      val id: String = split(0)
      //将学号作为key，成绩信息作为value
      (id,score)
    })

    //使用内连接关联数据
    val interJoinRDD: RDD[(String, (String, String))] = stuKvRDD.join(scoKvRDD)

    //interJoinRDD.foreach(println)
    //处理连接表格得到想得到的数据
   val resultRDD: RDD[(String, String)] = interJoinRDD.map{
      case (_:String,(stuInfo:String,scoInfo:String))=>
        val name: String = stuInfo.split(",")(1)
        val score: String = scoInfo.split(",")(2)
        (name,score)
    }


    /**
      * leftOutJoin:以左边为基础，如果右边没有用null代替
      */

    val leftOuterJoinRDD: RDD[(String, (String, Option[String]))] = stuKvRDD.leftOuterJoin(scoKvRDD)
    leftOuterJoinRDD.sample(false,0.01).foreach(println)

    //处理关联数据
    val leftOuterJoinResultRDD: RDD[String] =leftOuterJoinRDD.map{
      //关联上的处理方法
      case (_:String,(stuInfo:String,Some(scoInfo)))
      =>stuInfo+"\t"+scoInfo
      //没有关联上的处理方法
      case(_:String,(stuInfo:String,None))
      =>stuInfo+"\t"+"默认"
    }
    leftOuterJoinResultRDD.sample(false,0.01).foreach(println)

    /**
      * fullOuterJoin
      */
    val fullOuterJoinRDD: RDD[(String, (Option[String], Option[String]))] = stuKvRDD.fullOuterJoin(scoKvRDD)

    val fullOuterJoinResultRDD: RDD[String] = fullOuterJoinRDD.map{
      //关联上的处理方法
      case (id:String,(Some(stuInfo),Some(scoInfo))) =>stuInfo+"\t"+scoInfo

      //没有关联上的处理方法
      case (id:String,(None,Some(scoInfo))) =>"默认值"+"\t"+scoInfo

      case (id:String,(Some(stuInfo),None)) =>stuInfo+"\t"+"默认值"
    }
    fullOuterJoinResultRDD.sample(false,0.01).foreach(println)
  }
}
